package com.atguigu.flink.chapter10.query;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/29 09:09
 */
public class TopN {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        
        tEnv.executeSql("create table ub(" +
                            " user_id bigint, " +
                            " item_id bigint, " +
                            " category_id bigint, " +
                            " behavior string, " +
                            " ts bigint,  " +
                            " et as to_timestamp_ltz(ts, 0), " +
                            " watermark for et as et - interval '3' second " +
                            ")with(" +
                            "  'connector' = 'filesystem', " +
                            "  'path' = 'input/UserBehavior.csv', " +
                            "  'format' = 'csv' " +
                            ")");
        
        // 1. 统计每个在每个窗口的点击量
        Table t1 = tEnv.sqlQuery("select " +
                                     " item_id, window_end, count(*) ct " +
                                     "from table( tumble( table ub, descriptor(et), interval '1' hour) ) " +
                                     "where behavior='pv' " +
                                     "group by item_id, window_start, window_end");
        tEnv.createTemporaryView("t1", t1);
        // 2. 按照点击量在每个窗口进行降序, 给每个点击量一个排名
        Table t2 = tEnv.sqlQuery("select " +
                                     " item_id, window_end, ct, " +
                                     " row_number() over(partition by window_end order by ct desc) rn " +
                                     "from t1");
        tEnv.createTemporaryView("t2", t2);
        
        
        // 3. 过滤出 topN where rn<=3
        Table result = tEnv.sqlQuery("select * " +
                                         "from t2 " +
                                         "where rn<= 3");
        
        
        tEnv.executeSql("create table ub_topn( " +
                            "    item_id bigint, " +
                            "    window_end timestamp(3), " +
                            "    ct bigint, " +
                            "    rn bigint," +
                            "    primary key(window_end,rn) not enforced " +
                            ")with(" +
                            " 'connector' = 'jdbc'," +
                            " 'url' = 'jdbc:mysql://hadoop162:3306/flink?useSSL=false', " +
                            " 'table-name' = 'ub_topn', " +
                            " 'username' = 'root', " +
                            " 'password' = 'aaaaaa' " +
                            ")");
    
        result.executeInsert("ub_topn");
        
        
        
    }
}
/*
mysql 建表语句:
create table ub_topn(
    item_id bigint,
    window_end timestamp,
    ct bigint,
    rn bigint,
    primary key(window_end,rn)
)
 */